package com.cnlive.logpare.play

import java.text.SimpleDateFormat
import java.util.Locale
import org.apache.spark.SparkConf
import com.cnlive.logpare.util.Constant
import org.apache.spark.SparkContext
import org.apache.spark.sql.Row
import java.net.URLDecoder
import java.net.URL
import com.cnlive.logpare.util.ParseLogUtil
import scala.collection.mutable.HashMap
import scala.collection.immutable.Map
import com.cnlive.logpare.util.IPSeekerExt

object ParseNginxPlayLog {
    val dateFormat=new SimpleDateFormat("yyyyMMddHHmmss",Locale.ENGLISH)
    val dateFormat2=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss",Locale.ENGLISH)
    def main(args: Array[String]): Unit = {
       var env=1 //运行环境 1.本地环境，2，本地测试环境.3远程测试环境
       if(args.length>0){
         env=args(0).toInt 
       }
       val files=env match{
            case 1 => "file:///F:/工作资料/openapi_access_201706220215.log"
            case _ => "hdfs://sxt/nginx_logs/app/offline/2017-06-15/*"
       }
       val qqwryPath=env match{
                case 2 =>Constant.qqwryLocalServer
                case 3 =>Constant.qqwryServer
                case _=>""
       }
        val conf = new SparkConf().setAppName("ParsePlayLog")
        if(env==1){
          conf.setMaster("local")
        }
      
        val sc = new SparkContext(conf)
        sc.textFile(files).mapPartitions(x=>{
           var result = List[Row]()
           val iPSeekerExt=env match {
              case 1 =>  IPSeekerExt.getInit()
              case 2 =>  IPSeekerExt.getInit(true,qqwryPath)
              case 3 =>  IPSeekerExt.getInit(true,qqwryPath)
           }
           while (x.hasNext) {
             val textLine=x.next()
             if(textLine.indexOf("playBy") != -1){
                 parLineText(iPSeekerExt,textLine)
             }
           }
           result.iterator
        }).count()
    }
/*(1,303)//("live","H5")
(1,301)//("live","SWF")
(1,302)//("live","APP")
(2,303)//("vod","H5")
(2,301)//("vod","SWF")
(2,302)//("vod","APP")

*/  val listPalyMap=Map("liveplayByH5"->(1,303),
                        "liveplayBySWF"->(1,301),
                       "liveplayByAPP"->(1,302),
                       "vodSWF"->(1,301),
                       "vodplayByH5"->(1,303),
                       "vodplayByAPP"->(1,302)
    )
    val listPlayStr=List("liveplayByH5",
               "liveplayBySWF",
               "liveplayByAPP",
               "vodSWF",
               "vodplayByH5",
               "vodplayByAPP")
    val ts=""" \"(.*?)\"""".r
    def parLineText(iPSeekerExt:IPSeekerExt,lineText:String){
        val  mapRequest=new HashMap[String, String]()
        val ipandtime=ts.split(lineText).toList
        val urlandtype=ts.findAllIn(lineText).toList
        if(ipandtime.length==2&&urlandtype.length==3){
          val ipandtimeArr=ipandtime(0).split(" ")//由于IP可能有多个所以
          val timeIndx=ipandtimeArr.length-2
          val ip=if (ipandtimeArr(0).indexOf(",") != -1) ipandtimeArr(0).substring(0, ipandtimeArr(0).indexOf(",")) else ipandtimeArr(0)
          ParseLogUtil.parseIptoLocation(iPSeekerExt, ip, mapRequest)//解析ip地址--省份--运营商
          val time=ipandtimeArr(timeIndx).replace("[", "")
          val urlandper=urlandtype(0).split(" ")
          if(urlandper.length==4){
            ParseLogUtil.parseUrlToMap(urlandper(2), mapRequest)//解析url中的参数
            for(playTpye<-listPalyMap.keySet){
              if(urlandper(2).indexOf(playTpye) != -1){
                val typef=listPalyMap.getOrElse(playTpye, (0,399))
                mapRequest.put("playtype", typef._1+"");
                mapRequest.put("actitem", typef._2+"");
              }
            }
          }else{
            println("url不合格"+urlandtype)
          }
        }else{
          println("数据格式不对--"+lineText)
        }
        println(mapRequest)
    }
}